// Copyright (C) 2012 Yury Bychenkov (http://bychenkov.com).
// All rights reserved.
//
// Redistribution and use in source and binary forms, with or without
// modification, are permitted provided that the following conditions are
// met:
//
//  - Redistributions of source code must retain the above copyright
// notice, this list of conditions and the following disclaimer.
//  - Redistributions in binary form must reproduce the above
// copyright notice, this list of conditions and the following disclaimer
// in the documentation and/or other materials provided with the
// distribution.
//  - Neither the name of Yury Bychenkov nor the names of its
// contributors may be used to endorse or promote products derived from
// this software without specific prior written permission.
//
// THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
// "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
// LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
// A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
// OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
// SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
// LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
// DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
// THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
// (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
// OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.

inline uint NunChunk::getColumnCount() const {
	return header->columnCount;
}

inline const Dictionary& NunChunk::getDictionary(uint columnIndex) const {
	return dictionaries[columnIndex];
}

inline const Index& NunChunk::getIndex(uint columnIndex) const {
	return indecies[columnIndex];
}

inline uint NunChunk::getType(uint columnIndex) const {
	return header->columns[columnIndex].type;
}



inline NunChunkProvider::NunChunkProvider(const Schema& schema_, const NunChunk& chunk_): schema(schema_), chunk(chunk_) {
}

inline const Schema& NunChunkProvider::getSchema() const {
	return schema;
}

inline uint NunChunkProvider::getCardinality(uint columnIndex) const {
	return chunk.getDictionary(columnIndex).getCount();
}


template <typename ElementTraits> 
int NunChunkProvider::find(uint columnIndex, const uint8* value) const {
	return chunk.getDictionary(columnIndex).find<ElementTraits>(value);
}

inline bool NunChunkProvider::isIndexAvailable(uint columnIndex) const {
	return true;
}

inline NunChunkProvider::Cursor::Cursor(const NunChunkProvider& provider_, const uint* columnMapping, uint columnCount_, const uint32* rangeFilter, uint rangeFilterCount):
	valueIndex(0),
	columnCount(columnCount_),
	provider(provider_) 
{
	for (uint i = 0; i < columnCount; i++) {
		if (i < rangeFilterCount) {
			iterators[i] = provider.chunk.getIndex(columnMapping[i]).iterator(rangeFilter[2 * i], rangeFilter[2 * i + 1]);
		} else {
			iterators[i] = provider.chunk.getIndex(columnMapping[i]).iterator();
		}
		dictionaries[i] = &provider.chunk.getDictionary(columnMapping[i]);
	}
}

inline uint32 NunChunkProvider::Cursor::getIndex(uint columnIndex) {
	Index::Iterator& it = iterators[columnIndex];
	uint skipCount = valueIndex - it.getValueIndex();
	if (skipCount > 0) {
		it.skip(skipCount);
	}
	return it.getValue();
}

inline const uint8* NunChunkProvider::Cursor::getString(uint columnIndex) {
	return dictionaries[columnIndex]->get<StringTraits>(getIndex(columnIndex));
}

inline const uint8* NunChunkProvider::Cursor::getDecimal(uint columnIndex) {
	return dictionaries[columnIndex]->get<DecimalTraits>(getIndex(columnIndex));
}

inline const uint8* NunChunkProvider::Cursor::getInteger(uint columnIndex) {
	return dictionaries[columnIndex]->get<IntegerTraits>(getIndex(columnIndex));
}


inline uint NunChunkProvider::Cursor::getValueIndex() const {
	return valueIndex;
}

inline bool NunChunkProvider::Cursor::next() {
	if (!iterators[0].skip()) {
		return false;
	}
    valueIndex = iterators[0].getValueIndex();

    const uint prefetchColumnCount = columnCount;

  	for (uint columnIndex = 1; columnIndex < prefetchColumnCount; ) {
		Index::Iterator& it = iterators[columnIndex];
  		if (!it.skip(valueIndex - it.getValueIndex())) {
  			return false;
  		}
  		if (valueIndex < it.getValueIndex()) {
  			valueIndex = it.getValueIndex();
  		    columnIndex = columnIndex == 0 ? 1 : 0;
  		} else {
  			columnIndex++;
  		}
	}

	return true;
}
